package com.demo.kafka;

import java.util.Properties;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * kafka发布者
 *
 */
public class KafkaPublisher {
	//初始化logger
	private static final Logger logger = LoggerFactory.getLogger(KafkaPublisher.class);
	//地址
	private String address;
	//重试
	private int retries;
	//批量长度
	private int batchSize;
	//
	private String acks;
	//缓冲存储
	private long bufferMemory;
	//串行键
	private String keySerializer;
	//串行值
	private String valueSerializer;
	//属性
	private Properties props;
	
	
	/**
	 * 无参构造
	 */
	public KafkaPublisher() {
		
	}

	/**
	 * 获得address
	 * @return the address
	 */
	public String getAddress() {
		return address;
	}

	/**
	 * 设置address
	 * @param address the address to set
	 */
	public void setAddress(String address) {
		this.address = address;
	}

	/**
	 * 获得retries
	 * @return the retries
	 */
	public int getRetries() {
		return retries;
	}

	/**
	 * 设置retries
	 * @param retries the retries to set
	 */
	public void setRetries(int retries) {
		this.retries = retries;
	}

	/**
	 * 获得batchSize
	 * @return the batchSize
	 */
	public int getBatchSize() {
		return batchSize;
	}

	/**
	 * 设置batchSize
	 * @param batchSize the batchSize to set
	 */
	public void setBatchSize(int batchSize) {
		this.batchSize = batchSize;
	}

	/**
	 * 获得props
	 * @return the props
	 */
	public Properties getProps() {
		return props;
	}

	/**
	 * 设置props
	 * @param props the props to set
	 */
	public void setProps(Properties props) {
		this.props = props;
	}
	
	/**
	 * 获得acks
	 * @return the acks
	 */
	public String getAcks() {
		return acks;
	}

	/**
	 * 设置zcks
	 * @param acks the acks to set
	 */
	public void setAcks(String acks) {
		this.acks = acks;
	}

	/**
	 * 获得bufferMemory
	 * @return the bufferMemory
	 */
	public long getBufferMemory() {
		return bufferMemory;
	}

	/**
	 * 设置bufferMemory
	 * @param bufferMemory the bufferMemory to set
	 */
	public void setBufferMemory(long bufferMemory) {
		this.bufferMemory = bufferMemory;
	}

	/**
	 * 获得keySerializer
	 * @return the keySerializer
	 */
	public String getKeySerializer() {
		return keySerializer;
	}

	/**
	 * 设置keySerializer
	 * @param keySerializer the keySerializer to set
	 */
	public void setKeySerializer(String keySerializer) {
		this.keySerializer = keySerializer;
	}

	/**
	 * 获得valueSerializer
	 * @return the valueSerializer
	 */
	public String getValueSerializer() {
		return valueSerializer;
	}

	/**
	 * 设置valueSerializer
	 * @param valueSerializer the valueSerializer to set
	 */
	public void setValueSerializer(String valueSerializer) {
		this.valueSerializer = valueSerializer;
	}

	/**
	 * 发布
	 * @param topic
	 * @param key
	 * @param value
	 */
	public void publish(String topic, String key, byte[] value) {
		Producer<String, byte[]> producer = null;
		
		initializeProps();
		
		try {
			producer = new KafkaProducer<>(props);
			ProducerRecord<String, byte[]> record = new ProducerRecord<String, byte[]>(topic, key, value);
			
			producer.send(record);
		} catch (Exception e) {
			logger.error("publish error: " + e);
			throw new RuntimeException("publish topic = " + topic + " error: " + e);
		} finally {
			if (producer != null) {
				producer.close();
			}
		}
	}
	
	/**
	 * 初始化属性
	 */
	public void initializeProps() {
		if (null == this.props) {
			this.props = new Properties();
			this.props.put("bootstrap.servers", this.address);
			this.props.put("retries", this.retries);
			this.props.put("batch.size", this.batchSize);
			this.props.put("acks", this.acks);
			this.props.put("buffer.memory", this.bufferMemory);
			this.props.put("key.serializer", this.keySerializer);
			this.props.put("value.serializer", this.valueSerializer);
		}
	}
}
